package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo21Student {
  def main(args: Array[String]): Unit = {
    // 5、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数]
    // 归一化->平均值->方差->排序

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("Demo21Student")

    val sc: SparkContext = new SparkContext(conf)

    // 读取学生数据构建RDD
    val stuRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    val scoRDD: RDD[String] = sc.textFile("spark/data/score.txt")


    val subRDD: RDD[String] = sc.textFile("spark/data/subject.txt")

    val scoKVRDD: RDD[(String, (String, Int))] = scoRDD.map(line => {
      val splits: Array[String] = line.split(",")
      val stuId: String = splits(0)
      val subId: String = splits(1)
      val score: Int = splits(2).toInt
      (stuId, (subId, score))
    })
    // 数据比较小，可以广播
    val subKVRDD: RDD[(String, (String, String))] = subRDD.map(line => {
      val splits: Array[String] = line.split(",")
      val subID: String = splits(0)
      val subName: String = splits(1)
      val subScore: String = splits(2)
      (subID, (subName, subScore))
    })
    val subKVMap: collection.Map[String, (String, String)] = subKVRDD.collectAsMap()
    val subBroMap: Broadcast[collection.Map[String, (String, String)]] = sc.broadcast(subKVMap)

    // map join
    // 归一化
    val standScoreRDD: RDD[(String, (String, Double))] = scoKVRDD.map {
      case (stuId: String, (subId: String, score: Int)) => {
        val subScore: Double = subBroMap.value(subId)._2.toDouble
        (stuId, (subId, score * 100 / subScore))
      }
    }
    // 平均分
    val avgScoreRDD: RDD[(String, Double)] = standScoreRDD.groupBy(_._1).map(kv => {
      val stuID: String = kv._1
      val scores: Iterable[Double] = kv._2.map(_._2).map(_._2)
      val sum_score: Double = scores.sum
      val score_cnt: Double = scores.size
      (stuID, sum_score / score_cnt)
    })

    val varianceRDD: RDD[(String, Double)] = standScoreRDD.join(avgScoreRDD)
      // 计算每个分数与平均分的差值的平方
      .map {
        case (stuId: String, ((subId: String, standScore: Double), avg_score: Double)) => {
          (stuId, Math.pow(standScore - avg_score, 2))
        }
      }.groupBy(_._1)
      // 计算方差
      .map(kv => {
        val stuID: String = kv._1
        val tmp: Iterable[Double] = kv._2.map(_._2)
        val variance: Double = tmp.sum / tmp.size
        (stuID, variance)
      })
    // 取出偏科最严重的前100名学生
    val stuIDsArr: Array[(String, Double)] = varianceRDD.sortBy(-_._2).take(100)
    stuIDsArr.foreach(println)

    // 根据100名学生的id去过滤、关联 得到最后的结果

  }

}
